package com.ruoyi.iot;

import java.util.Map;

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.TypeReference;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.iotcommon.constant.RocketTopicConstant;
import com.ruoyi.iotcommon.domain.DeviceDataMessage;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

/**
 * 设备上报消息处理器
 */
@Slf4j
@Component
public class MqttMessageHandler  implements MessageHandler{


    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     *  接收mqtt消息,alink协议 :/sys/${productKey}/${deviceKey}/thing/event/${channelKey}/post
     */
    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
      try {
        String topic = message.getHeaders().get("mqtt_receivedTopic", String.class);
        String[] topicArray = topic.split("/");
        String productKey = topicArray[2];
        String deviceKey = topicArray[3];
        String channelKey = topicArray[6];
        log.info("topic: {}",topic);
        log.info("message：{}",message.getPayload().toString());
        JSONObject msgBody = JSONObject.parseObject((String) message.getPayload());
  
        Map<String,Object> dataBody = msgBody.getObject("params", null );
        //  发送rocketmq
        DeviceDataMessage deviceDataMessage = new DeviceDataMessage();
        deviceDataMessage.setProductKey(productKey);
        deviceDataMessage.setDeviceKey(deviceKey);
        deviceDataMessage.setChannelKey(channelKey);
        deviceDataMessage.setTime(System.currentTimeMillis());
        deviceDataMessage.setParams(dataBody);
        rocketMQTemplate.sendOneWay(RocketTopicConstant.DEVICE_DATA, deviceDataMessage);
      } catch (Exception e) {
        log.error("处理mqtt消息失败",e);
      }

    }
    
}
